package com.kingsoft.dc.khaos.module.mpp.operator;

import com.kingsoft.dc.khaos.module.mpp.client.ConfVars;
import com.kingsoft.dc.khaos.module.mpp.client.KhaosContants;
import com.kingsoft.dc.khaos.module.mpp.client.MppClient;
import com.kingsoft.dc.khaos.module.mpp.exception.MppSqlException;
import com.kingsoft.dc.khaos.module.mpp.metadata.CombineStrategy;
import com.kingsoft.dc.khaos.module.mpp.metadata.MppConfig;
import com.kingsoft.dc.khaos.module.mpp.model.ColumnEntity;
import com.kingsoft.dc.khaos.module.mpp.sqlbuilder.GetAddSqlBuilder;
import com.kingsoft.dc.khaos.module.mpp.sqlbuilder.SqlBuilderUtils;
import com.kingsoft.dc.khaos.plugin.core.KhaosContext;
import org.apache.commons.lang3.StringUtils;

import java.util.List;

/**
 * 增量切片.
 */
public class TySnapshotAdd extends CombineStrategy {

    @Override
    public void execute(KhaosContext kc, String moduleId, MppConfig mppConfig) {
        //初始化MPP client,client已经初始化数据库连接
        MppClient client = null;
        try {
            client = new MppClient(mppConfig, kc);


            //获取操作的库表参数
            //比如获取源表所属的schema
            String srcSchema = String.format("\"%s\"", mppConfig.getSrcSchema());
            log().info("源库名srcSchema===>  " + srcSchema);
            //源表名
            String srcTable = String.format("\"%s\"", mppConfig.getSrcTable());
            log().info("源表名srcTable===>  " + srcTable);

            //源表过滤条件
            String filter = mppConfig.getDataFilter();

            //目标表所属schema名
            String tarSchema = String.format("\"%s\"", mppConfig.getTarSchema());
            log().info("目标库名tarSchema===>  " + tarSchema);
            //目标表名
            String tarTable = String.format("\"%s\"", mppConfig.getTarTable());
            log().info("目标表名tarTable===>  " + tarTable);

            //目标表名(上一业务日期)所属schema名
            String preSchema = String.format("\"%s\"", mppConfig.getPreSchema());
            log().info("上一业务日期目标库名preSchema===>  " + preSchema);
            //目标表名(上一业务日期)
            String preTable = String.format("\"%s\"", mppConfig.getPreTable());
            log().info("上一业务日期目标表名preTable===>  " + preTable);

            //减档表所属schema名
            String delSchema = String.format("\"%s\"", mppConfig.getDelSchema());
            log().info("减档库名delSchema===>  " + delSchema);
            //减档表名
            String delTableName = String.format("\"%s\"", mppConfig.getDelTable());
            log().info("减档表名delTableName===>  " + delTableName);

            //主键列表
            List<String> keys = mppConfig.getpKeys();

            //获取源档类型
            String src_Type = mppConfig.getSrcType();
            //获取主键字符串
            //String pKeys = SqlBuilderUtils.selKeys(tarTable,keys);
            //是否物理删除
            boolean isDeleted = mppConfig.isDeleted();
            //获取追加字段名


            //获取追加字段值，比如获取[数据日期]字段值
            String bizDate = kc.getConf().getString(KhaosContants.BIZ_DATE);
            String batchNo = kc.getConf().getString(KhaosContants.BATCH_NO);
            String jobName = kc.getConf().getString(KhaosContants.JOB_NAME);
            String jobId = kc.getConf().getString(KhaosContants.JOB_INSTANCE_ID);
            //获取表全字段信息如id,name
            List<String> columnEntities = client.getColumnNameList(mppConfig.getSrcSchema(), mppConfig.getSrcTable());
            log().info("源表全字段信息===>  " + StringUtils.join(columnEntities, ","));

            //获取表及表全字段信息
            List<ColumnEntity> tableColumns = client.getColumnInfo(mppConfig.getSrcSchema(), mppConfig.getSrcTable());
            //获取目标表主键字段信息
            String columns = SqlBuilderUtils.selKeys(tarTable, columnEntities);
            log().info("目标表主键字段信息===>  " + columns);

            //获取两表关联主键SQL字符串
            String pKs = SqlBuilderUtils.joinKeys(delTableName, tarTable, keys);

            //获取两表关联主键SQL字符串
            String pKst = SqlBuilderUtils.joinKeys(preTable, srcTable, keys);
            //获取两表关联字段空值处理SQL字符串
            String columnNulls = SqlBuilderUtils.replaceNull(srcTable, tarTable, tableColumns);
            //获取减档表别名及表全字段信息如s.id,s.name
            String selTabCols = SqlBuilderUtils.selTabCols(delTableName, columnEntities);
            log().info("减档表别名及表全字段信息===>  " + selTabCols);
            //获取表别名+表全字段信息如s.id,s.name
            //String selTabColst = SqlBuilderUtils.selTabCols(tarTable,columnEntities);
            //获取表别名+表全字段信息如s.id,s.name
            // String selTabColst1 = SqlBuilderUtils.selTabCols(tarTable + "1",columnEntities);

            //获取源档是否有记录数
            Boolean srcCount = client.hasRecord(srcSchema, srcTable);

            //求增量表、减量表
            String sql_add_record = srcSchema + "." + srcTable;         //源增量表
            String sql_del_record = delSchema + "." + delTableName;     //源减量表
            String preAllTable = preSchema + "." + preTable;            //上期全量表
            String tar_tbl_name = tarSchema + "." + tarTable;

            if (src_Type.equalsIgnoreCase("ALL")) {
                sql_add_record = " ( " + GetAddSqlBuilder.getAddSelectSql(sql_add_record, preAllTable, keys, tableColumns, filter) + " ) as d  ";
                sql_del_record = " ( " + GetAddSqlBuilder.getDelSelectSql(sql_add_record, preAllTable, keys, filter) + " ) as d  ";
                //获取两表关联主键SQL字符串
                pKs = SqlBuilderUtils.joinKeys(srcTable, tarTable, keys);
                delTableName = srcTable;
                selTabCols = SqlBuilderUtils.selTabCols(srcTable, columnEntities);
            }
            //支持重跑算子SQL
            String rerun_sql = rerunSql(tarSchema, tarTable, bizDate, batchNo, client);
//            System.out.println(rerun_sql);
            client.execInsert(rerun_sql);

            //拼接增量切片表全量档特殊算子SQL
            String exeu_sql = buildSqlall(src_Type, tarSchema, tarTable, srcTable, bizDate, jobName, batchNo, columns, isDeleted, selTabCols, sql_del_record, keys, pKs, delTableName, client);
//            System.out.println(exeu_sql);
            client.execInsert(exeu_sql);

            String exe_sql = "";
            //拼接增量切片表算子SQL
            if (null == filter || filter.equals("")) {
                exe_sql = buildSql(src_Type, tarSchema, tarTable, srcSchema, srcTable, bizDate, jobName, batchNo, columns, pKst, isDeleted, sql_add_record, preSchema, preTable, srcCount, client);
            } else {
                exe_sql = buildSqlWithFilter(src_Type, tarSchema, tarTable, srcSchema, srcTable, bizDate, jobName, batchNo, columns, pKst, isDeleted, sql_add_record, preSchema, preTable, srcCount, client, filter);
            }
            //            System.out.println(exe_sql);
            client.execInsert(exe_sql);
        } catch (Exception e) {
            e.printStackTrace();
            throw new MppSqlException("执行MPP SQL 异常");
        } finally {
            if (client != null) {
                client.close();
            }
        }
    }

    //重跑作业清理
    public String rerunSql(String tarSchema, String tarTable, String bizDate, String batchNo, MppClient client) {
        System.out.println("TySnapshotAdd build rerunSql start...");
        String tyDataDate = client.getAppendColName(ConfVars.P9_DATA_DATE);
        String tyBatchNumber = client.getAppendColName(ConfVars.P9_BATCH_NUMBER);
        //重跑作业清理
        String del_sql = String.format("\n BEGIN WORK; \n" +
                "\n  delete from %s.%s \n" +
                "  where " + tyDataDate + " = to_date('%s','YYYYMMDD')\n" +
                "    and " + tyBatchNumber + " = '%s';\n", tarSchema, tarTable, bizDate, batchNo);
        System.out.println("TySnapshotAdd build rerunSql end...");
        return del_sql;
    }

    //拼接增量切片表全量档算子SQL实现类
    public String buildSqlall(String src_Type,
                              String tarSchema,
                              String tarTable,
                              String srcTable,
                              String bizDate,
                              String jobName,
                              String batchNo,
                              String columns,
                              boolean isDeleted,
                              String selTabCols,
                              String sql_del_record,
                              List<String> keys,
                              String pKs,
                              String delTable,
                              MppClient client) {
        System.out.println("TySnapshotAdd buildSqlall start...");
        //比如获取[数据日期]字段名
        String tyDataDate = client.getAppendColName(ConfVars.P9_DATA_DATE);
        String tyBatchNumber = client.getAppendColName(ConfVars.P9_BATCH_NUMBER);
        String tyDelFlag = client.getAppendColName(ConfVars.P9_DEL_FLAG);
        String tyDelDate = client.getAppendColName(ConfVars.P9_DEL_DATE);
        String tyDelBatch = client.getAppendColName(ConfVars.P9_DEL_BATCH);
        String tyJobName = client.getAppendColName(ConfVars.P9_JOB_NAME);
        String all_sql = "";
        //全量有物理删除或者增量有物理删除，则需要把删除标记和删除日期置为原样,把如下SQL拼接到$job_clear_sql
        if ((src_Type.equalsIgnoreCase("ADD") && isDeleted) || (src_Type.equalsIgnoreCase("ALL") && isDeleted)) {
            String all_del_sql_one = String.format(" \n insert into %s.%s (\n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      " + tyDelFlag + ",\n" +
                    "\t      " + tyDelDate + ",\n" +
                    "\t      " + tyDelBatch + ",\n" +
                    "\t      " + tyJobName + "\t\n" +
                    "\t) select \n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      '0' as " + tyDelFlag + ",\n" +
                    "\t      to_date('29991231','YYYYMMDD') as " + tyDelDate + ",\n" +
                    "\t      '0000' as " + tyDelBatch + ",\n" +
                    "\t      " + tyJobName + " \n" +
                    "\t  from %s.%s \n" +
                    "\t  where  " + tyDelFlag + " = '1' \n" +
                    "\t     and  " + tyDelDate + " = to_date('%s','YYYYMMDD')\n" +
                    "\t     and  " + tyDelBatch + " = '%s'; \n", tarSchema, tarTable, columns, columns, tarSchema, tarTable, bizDate, batchNo);

            String all_del_sql_two = String.format("\n delete from %s.%s \n" +
                    "\t where " + tyDelFlag + " = '1'\n" +
                    "\t   and " + tyDelDate + " = to_date('%s','YYYYMMDD')\n" +
                    "\t   and " + tyDelBatch + " = '%s'; \n", tarSchema, tarTable, bizDate, batchNo);


            //全量需要分离增量数据和减量数据，默认全字段对比
            String all_del_sql_three = String.format(" \n insert into %s.%s (\n" +
                            "\t      %s,\n" +
                            "\t      " + tyDataDate + ",\n" +
                            "\t      " + tyBatchNumber + ",\n" +
                            "\t      " + tyDelFlag + ",\n" +
                            "\t      " + tyDelDate + ",\n" +
                            "\t      " + tyDelBatch + ",\n" +
                            "\t      " + tyJobName + "\t\n" +
                            "\t) select \n" +
                            "\t      %s,\n" +
                            "\t      %s." + tyDataDate + ",\n" +
                            "\t      %s." + tyBatchNumber + ",\n" +
                            "\t      '1' as " + tyDelFlag + ",\n" +
                            "\t      to_date('%s','YYYYMMDD') as " + tyDelDate + ",\n" +
                            "\t      '%s' as " + tyDelBatch + ",\n" +
                            "\t      '%s' as " + tyJobName + " \n" +
                            "\t  from %s.%s %s\n" +
                            "\t  left join (select * from  %s ) %s\n" +
                            "\t   on   %s \n" +
                            "\t  where %s." + tyDelFlag + " = '0' \n" +
                            "\t   and  %s.%s is not null; \n", tarSchema, tarTable, columns, selTabCols, tarTable, tarTable, bizDate, batchNo
                    , jobName, tarSchema, tarTable, tarTable, sql_del_record, delTable, pKs, tarTable, delTable, keys.get(0));
            //把已经置删除标记的记录从目标表移除
            String all_del_sql_four = String.format("delete from %s.%s %s \n" +
                    "\t      using (select * from %s ) %s \n" +
                    "\t      where %s \n" +
                    "\t        and %s." + tyDelFlag + " = '0'; \n", tarSchema, tarTable, tarTable, sql_del_record, delTable, pKs, tarTable);

            all_sql = all_sql + all_del_sql_one + all_del_sql_two + all_del_sql_three + all_del_sql_four;
        }
        System.out.println("TySnapshotAdd buildSqlall end...");
        return all_sql;
    }

    //拼接增量切片表增量减量全量档算子SQL实现类
    public String buildSqlWithFilter(String src_Type,
                                     String tarSchema,
                                     String tarTable,
                                     String srcSchema,
                                     String srcTable,
                                     String bizDate,
                                     String jobName,
                                     String batchNo,
                                     String columns,
                                     String pKst,
                                     boolean isDeleted,
                                     String sql_add_record,
                                     String preSchema,
                                     String preTable,
                                     Boolean srcCount,
                                     MppClient client,
                                     String filter) {
        System.out.println("TySnapshotAdd buildSqlWithFilter start...");
        String tyDataDate = client.getAppendColName(ConfVars.P9_DATA_DATE);
        String tyBatchNumber = client.getAppendColName(ConfVars.P9_BATCH_NUMBER);
        String tyDelFlag = client.getAppendColName(ConfVars.P9_DEL_FLAG);
        String tyDelDate = client.getAppendColName(ConfVars.P9_DEL_DATE);
        String tyDelBatch = client.getAppendColName(ConfVars.P9_DEL_BATCH);
        String tyJobName = client.getAppendColName(ConfVars.P9_JOB_NAME);
        String all_sql = "";

        String schemaTblFilter = String.format("(select * from %s.%s where %s)", srcSchema, srcTable, filter);

        //#1.源档是增量或者是2.全量需处理增量或者是3.源档是增量配且置物理删除
        if ((src_Type.equalsIgnoreCase("ALL")) || (src_Type.equalsIgnoreCase("ADD")) || (src_Type.equalsIgnoreCase("ADD") && isDeleted)) {
            String exe_sql_one = String.format("\n insert into %s.%s (\n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      " + tyDelFlag + ",\n" +
                    "\t      " + tyDelDate + ",\n" +
                    "\t      " + tyDelBatch + ",\n" +
                    "\t      " + tyJobName + "\t\n" +
                    "\t) select \n" +
                    "\t      %s,\n" +
                    "\t      to_date('%s','YYYYMMDD') as " + tyDataDate + ",\n" +
                    "\t      '%s' as " + tyBatchNumber + ",\n" +
                    "\t      '0' as " + tyDelFlag + ",\n" +
                    "\t\t    to_date('29991231','YYYYMMDD') as " + tyDelDate + ",\n" +
                    "\t\t    '0000' as " + tyDelBatch + ", \n" +
                    "\t      '%s' as " + tyJobName + " \n" +
                    "\t  from %s %s;\n", tarSchema, tarTable, columns, columns, bizDate, batchNo, jobName, schemaTblFilter, srcTable);

            all_sql = all_sql + exe_sql_one;

            //源档是全量档需更新T表
            if (src_Type.equalsIgnoreCase("ALL") && srcCount) {
                //全量有物理删除
                if (isDeleted) {
                    String exe_sql_two = String.format("\n delete from %s.%s ;\n", preSchema, preTable);
                    all_sql = all_sql + exe_sql_two;
                } else {
                    String exe_sql_three = String.format("\n delete from %s.%s %s \n using %s.%s %s \n where %s;\n", preSchema, preTable, preTable, srcSchema, srcTable, srcTable, pKst);
                    all_sql = all_sql + exe_sql_three;
                }
                String exe_sql_four = String.format("\n insert into %s.%s (\n" +
                        "\t      %s\n" +
                        "\t) select \n" +
                        "\t      %s\n" +
                        "\t  from %s %s;\n", preSchema, preTable, columns, columns, schemaTblFilter, srcTable);
                all_sql = all_sql + exe_sql_four;
            }
        }
        System.out.println("TySnapshotAdd buildSqlWithFilter end...");
        return all_sql + "\n COMMIT ; \n ";
    }

    //拼接增量切片表增量减量全量档算子SQL实现类
    public String buildSql(String src_Type,
                           String tarSchema,
                           String tarTable,
                           String srcSchema,
                           String srcTable,
                           String bizDate,
                           String jobName,
                           String batchNo,
                           String columns,
                           String pKst,
                           boolean isDeleted,
                           String sql_add_record,
                           String preSchema,
                           String preTable,
                           Boolean srcCount,
                           MppClient client) {
        System.out.println("TySnapshotAdd buildSql start...");
        String tyDataDate = client.getAppendColName(ConfVars.P9_DATA_DATE);
        String tyBatchNumber = client.getAppendColName(ConfVars.P9_BATCH_NUMBER);
        String tyDelFlag = client.getAppendColName(ConfVars.P9_DEL_FLAG);
        String tyDelDate = client.getAppendColName(ConfVars.P9_DEL_DATE);
        String tyDelBatch = client.getAppendColName(ConfVars.P9_DEL_BATCH);
        String tyJobName = client.getAppendColName(ConfVars.P9_JOB_NAME);
        String all_sql = "";
        //#1.源档是增量或者是2.全量需处理增量或者是3.源档是增量配且置物理删除
        if ((src_Type.equalsIgnoreCase("ALL")) || (src_Type.equalsIgnoreCase("ADD")) || (src_Type.equalsIgnoreCase("ADD") && isDeleted)) {
            String exe_sql_one = String.format("\n insert into %s.%s (\n" +
                    "\t      %s,\n" +
                    "\t      " + tyDataDate + ",\n" +
                    "\t      " + tyBatchNumber + ",\n" +
                    "\t      " + tyDelFlag + ",\n" +
                    "\t      " + tyDelDate + ",\n" +
                    "\t      " + tyDelBatch + ",\n" +
                    "\t      " + tyJobName + "\t\n" +
                    "\t) select \n" +
                    "\t      %s,\n" +
                    "\t      to_date('%s','YYYYMMDD') as " + tyDataDate + ",\n" +
                    "\t      '%s' as " + tyBatchNumber + ",\n" +
                    "\t      '0' as " + tyDelFlag + ",\n" +
                    "\t\t    to_date('29991231','YYYYMMDD') as " + tyDelDate + ",\n" +
                    "\t\t    '0000' as " + tyDelBatch + ", \n" +
                    "\t      '%s' as " + tyJobName + " \n" +
                    "\t  from %s;\n", tarSchema, tarTable, columns, columns, bizDate, batchNo, jobName, sql_add_record);

            all_sql = all_sql + exe_sql_one;

            //源档是全量档需更新T表
            if (src_Type.equalsIgnoreCase("ALL") && srcCount) {
                //全量有物理删除
                if (isDeleted) {
                    String exe_sql_two = String.format("\n delete from %s.%s ;\n", preSchema, preTable);
                    all_sql = all_sql + exe_sql_two;
                } else {
                    String exe_sql_three = String.format("\n delete from %s.%s %s \n using %s.%s %s \n where %s;\n", preSchema, preTable, preTable, srcSchema, srcTable, srcTable, pKst);
                    all_sql = all_sql + exe_sql_three;
                }
                String exe_sql_four = String.format("\n insert into %s.%s (\n" +
                        "\t      %s\n" +
                        "\t) select \n" +
                        "\t      %s\n" +
                        "\t  from %s.%s ;\n", preSchema, preTable, columns, columns, srcSchema, srcTable);
                all_sql = all_sql + exe_sql_four;
            }
        }
        System.out.println("TySnapshotAdd buildSql end...");
        return all_sql + "\n COMMIT ; \n ";
    }
}
